/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package edu.isi.pegasus.planner.subworkflow.estimator.implementation;

import edu.isi.pegasus.planner.classes.ADag;
import edu.isi.pegasus.planner.classes.PegasusBag;
import edu.isi.pegasus.planner.dax.ADAG;
import edu.isi.pegasus.planner.subworkflow.estimator.Estimator;
import edu.isi.pegasus.planner.subworkflow.partitioner.DAXAbstraction;
import edu.isi.pegasus.planner.subworkflow.partitioner.SiteAbstraction;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.Random;

/**
 *
 * @author wchen
 */
public class HEFT extends Estimator{

        //maybe you should write this part to Estimator
    private ADag subDag;
    private edu.isi.pegasus.planner.classes.Job headNode;
    private edu.isi.pegasus.planner.classes.Job rearNode;
 //   private Map pathMap;

    private Map weight;
    private Map AFT;
    private Map avail;
    private Map EST;
    private Map EFT;
    private int numOfslot;
    private Map job2rank;
    private Map rank2job;
    private double casedEFTTime;
    //private Map navigated;
    public HEFT(DAXAbstraction bigDax, SiteAbstraction site, ADag mDag, PegasusBag mBag, HashMap mJob2Time)
    {
        super(bigDax, site, mDag, mBag, mJob2Time);
        headNode = new edu.isi.pegasus.planner.classes.Job();
        headNode.jobID = "HEAD";
        
        rearNode = new edu.isi.pegasus.planner.classes.Job();
        rearNode.jobID = "REAR";
//        pathMap = new HashMap<String, Double>();
        weight  = new HashMap<Double, Double>();
        AFT     = new HashMap<Double, Double>();
        avail   = new HashMap<Integer, Double>();
        EST     = new HashMap<Integer, Map>();
        EFT     = new HashMap<Integer, Double>();
        //numOfslot = site.slot;
//        rank2job = new HashMap<Double,
//                edu.isi.pegasus.planner.classes.Job>();
//        job2rank = new HashMap<
//                edu.isi.pegasus.planner.classes.Job, Double>();
        rank2job = new HashMap<Double, String>();
        job2rank = new HashMap<String, Double>();

         subDag = bigDax.subDag;
         //this is same for each site;
         //addNode();
         /*Warning*/
         addNodeParallel();

         //navigated = rank2job.descendingMap();
         
        
    }


    //private double getRank(edu.isi.pegasus.planner.classes.Job job)
    private double getRank(String jobID)
    {
        if(job2rank.containsKey(jobID))
        {
            return (Double)job2rank.get(jobID);
        }
        if(!mJob2Time.containsKey(jobID))
            mJob2Time.put(jobID, 0.0);
                //(String)job.vdsNS.get(Pegasus.RUNTIME_KEY);
        double runtime = (Double)mJob2Time.get(jobID);
        double rank;
        double runtime1 = 0.0;
        for(Iterator it = subDag.getChildren(jobID).iterator();it.hasNext();)
        {
//            edu.isi.pegasus.planner.classes.Job child = (edu.isi.pegasus.planner.classes.Job)
//                    it.next();
            String child = (String)it.next();
            double runtime2 = getRank(child);
            if(runtime2 > runtime1)
                runtime1 = runtime2;
        }

        rank = runtime + runtime1;
        Random generator = new Random();
        while(rank2job.containsKey(rank))
        {
            rank += generator.nextDouble() / 100;
        }
        job2rank.put(jobID, rank);
        rank2job.put(rank, jobID);
        weight.put(rank, runtime);
        return rank;

    }
    public void clean()
    {
        //weight.clear();
        AFT.clear();
        avail.clear();
        EST.clear();
        EFT.clear();

        //job2rank.clear();
        //rank2job.clear();
        //navigated.clear();
    }
    private void addNode()
    {
        subDag.add(headNode);
        subDag.add(rearNode);
        for(Iterator it = subDag.jobIterator(); it.hasNext();)
        {
            edu.isi.pegasus.planner.classes.Job job =
                    (edu.isi.pegasus.planner.classes.Job)it.next();
            

            if(subDag.getChildren(job.jobID).size() == 0 && !job.jobID.equals(rearNode.jobID))
                subDag.addNewRelation(job.jobID, rearNode.jobID);
            if(subDag.getParents(job.jobID).size() == 0 && !job.jobID.equals(headNode.jobID))
                subDag.addNewRelation(headNode.jobID, job.jobID);
        }
        getRank(headNode.jobID);
    }
        private void addNodeParallel()
    {
        subDag.add(headNode);
        subDag.add(rearNode);
        int num  = 8;
        int length = subDag.getNoOfJobs();
        ArrayList threadList = new ArrayList();
        for(int i = 0; i < num; i ++)
        {
           threadList.add(new addFakeNode(subDag, i*length/num,(i+1)*length/num,headNode,rearNode));
        }
        for(int i = 0; i < num; i ++)
        {
            addFakeNode afn = (addFakeNode)threadList.get(i);
            afn.getDone();
        }
         long startTime = System.currentTimeMillis();

        getRank(headNode.jobID);
        //getRankInParallel();
         long endTime = System.currentTimeMillis();
         System.out.println("Duration for getrank: " + (endTime - startTime));

    }



    private double getAFT(double rank)
    {
//        edu.isi.pegasus.planner.classes.Job job =
//                (edu.isi.pegasus.planner.classes.Job)rank2job.get(rank);
        if(AFT.containsKey(rank))
            return (Double)AFT.get(rank);
        double runtime1 = 1e8;//big enough
        int slot = -1;
        for(int i = 0; i < numOfslot; i ++)
        {
            double runtime = (Double)weight.get(rank);

            double runtime2  = runtime + getEST(rank, i);
            if(runtime2 >= 1e8)
                System.err.println("Runtime of this job is too large\n");
            if(runtime2 < runtime1)
            {
                runtime1 = runtime2;
                slot = i;
            }

        }
        //double ava = getavail(slot);
        avail.put(slot, runtime1);
        AFT.put(rank, runtime1);
        
        return runtime1;
    }
    private void getRankInParallel()
    {
        int num2job = 4;
        ArrayList threads = new ArrayList();
        //pathMap.put(rearNode.jobID, 0.0);//critical
        LinkedList mQueue = new LinkedList<String>();
        mQueue.add(rearNode.jobID);

        for(int i = 0; i < num2job; i ++)
        {
            threads.add(new getRankParallel(subDag, job2rank, i, num2job, mJob2Time, mQueue, rank2job, weight));
        }
        for(int i = 0; i < num2job; i ++)
        {
            getRankParallel gcp = (getRankParallel)threads.get(i);
            gcp.getDone();
        }

    }
    private double getEST(double rank, int slot)
    {

//        edu.isi.pegasus.planner.classes.Job job =
//                (edu.isi.pegasus.planner.classes.Job)rank2job.get(rank);
        String jobID = (String)rank2job.get(rank);
        if(EST.containsKey(rank) && EST.get(rank)!=null)
        {
            Map est = (Map)EST.get(rank);
            if(est.get(slot)!= null)
            {
                return (Double)est.get(slot);
            }
        }
        if(!EST.containsKey(rank))
        {
            Map list = new HashMap<Integer,Double>();
            EST.put(rank, list);

        }
        Map list = (Map)EST.get(rank);
        double value = 0.0;
        if(!list.containsKey(slot))
        {
            double maxAFT = getavail(slot);
            for(Iterator it = subDag.getParents(jobID).iterator(); it.hasNext();)
            {
//                edu.isi.pegasus.planner.classes.Job parent =
//                (edu.isi.pegasus.planner.classes.Job)it.next();
                String parent = (String)it.next();
                double prank = (Double)job2rank.get(parent);
                double aft = getAFT(prank);
                if(aft > maxAFT) maxAFT = aft;
            }
            value = maxAFT;
            list.put(slot, value);
        }
        return value;
    }

    private double getavail(int slot)
    {
        if(!avail.containsKey(slot))return 0.0;
        double ava = (Double)avail.get(slot);
        return ava;
    }


    public double getEstimatedRuntime(SiteAbstraction newsite)
    {

        ADAG dag = bigDax.adag;
//        System.out.println(dag.mName + " has " + bigDax.subDag.getNoOfJobs());
        //for debugging. 35GB
//        if(dag.mName.equals("subworkflow0"))return 228.995;
//        if(dag.mName.equals("subworkflow1"))return 6625.531;
//        if(dag.mName.equals("subworkflow2"))return 4660.43;
//        if(dag.mName.equals("subworkflow0"))return 228.995;
//        if(dag.mName.equals("subworkflow1"))
//            if(newsite.slot==20)return 3166;
//            else
//                return 11211;
//        if(dag.mName.equals("subworkflow2"))
//            if(newsite.slot ==20)return 2226;
//            else
//                return 8123;
        //for debugging 20GB. 
        clean();
        if(numOfslot == newsite.slot)
        {
            return casedEFTTime;
        }
        numOfslot = newsite.slot;
        double rank1 = (Double)job2rank.get(headNode.jobID);
        double rank2 = (Double)job2rank.get(rearNode.jobID);

        double makespan  = getAFT(rank2);
        casedEFTTime = makespan;
        return makespan;

    }

}
